package org.onion.pattern.event.springboot.rocketmq.publish;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.onion.pattern.event.outbox.data.OutboxEventInfo;
import org.onion.pattern.event.publish.EventPublishChannel;
import org.onion.pattern.event.spring.listener.EventListenerHandler;
import org.onion.pattern.event.spring.outbox.OutboxEventConverterBundle;
import org.onion.pattern.event.springboot.rocketmq.annotation.RocketMQEvent;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQ;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQEventProperties;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQMessageProperties;
import org.onion.pattern.event.springboot.rocketmq.exception.RocketMQEventException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.stereotype.Component;


import java.text.MessageFormat;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;

@Component
@Slf4j
public class RocketMQEventPublishChannel implements EventPublishChannel , EventListenerHandler {

    private final DefaultMQProducer          producer;
    private final MessageQueueSelector       messageQueueSelector = new SelectMessageQueueByHash();
    private final RocketMQEventProperties    properties;
    @Autowired
    private       OutboxEventConverterBundle outboxEventConverterBundle;


    @Autowired
    public RocketMQEventPublishChannel(RocketMQEventProperties properties) {
        this.properties = properties;
        this.producer = RocketMQProducerProvider.createProducer(properties);
    }

    @SneakyThrows
    @PostConstruct
    public void init() {
        producer.start();
    }

    @PreDestroy
    public void shutdown() {
        if (this.producer != null) {
            this.producer.shutdown();
        }
    }

    @Override
    public boolean support(Object obj) {

        if ((obj instanceof OutboxEventInfo)) {
            OutboxEventInfo event = (OutboxEventInfo) obj;
            if (RocketMQ.CHANNEL_NAME.equalsIgnoreCase(event.getChannel())) {
                return true;
            }
            return false;
        }

        RocketMQEvent rocketMQEvent = AnnotationUtils.findAnnotation(obj.getClass(), RocketMQEvent.class);
        return rocketMQEvent != null;
    }

    @Override
    public void handle(Object event) {
        send(event);
    }

    @Override
    public void send(Object event) {
        OutboxEventInfo outboxEventInfo;
        if (event instanceof OutboxEventInfo) {
            outboxEventInfo = (OutboxEventInfo) event;
        } else {
            outboxEventInfo = outboxEventConverterBundle.convert(event);    //带有注解的
        }

        publish(outboxEventInfo);
    }

    @Override
    public void send(List<Object> events) {
        for (Object event : events) {
            send(event);
        }
    }

    @Override
    public int getOrder() {
        return 100;
    }


    protected void publish(OutboxEventInfo event) {
        if (event.getPayload() == null || event.getPayload().length <= 0) {
            log.error("【事件】消息发送没有内容 {}", event);
            return;
        }

        Objects.requireNonNull(event.getPayload());
        Objects.requireNonNull(event.getTopic());
        Objects.requireNonNull(event.getEventKey());
        String hashKey = event.getDomainType() + "-" + event.getDomainId();
        SendResult sendResult = null;
        long publishBegin = System.currentTimeMillis();
        try {
            Message rocketMsg = createMessage(event.getTopic(), event.getSubTopic(), event.getPayload());
            if (event.getProps() != null && !event.getProps().isEmpty()) {
                final Map<String, String> properties = event.getProps();
                for (String key : properties.keySet()) {
                    rocketMsg.getProperties().put(key, properties.get(key));
                }
            }
//      MessageAccessor.putProperty(rocketMsg, "__SHARDINGKEY", hashKey);
            rocketMsg.getProperties().put(RocketMQMessageProperties.PAYLOAD_CLASS_NAME, event.getDomainType());
            rocketMsg.getProperties().put(RocketMQMessageProperties.PAYLOAD_SERIALIZER_TYPE, event.getSerializer());
            rocketMsg.getProperties().put(RocketMQMessageProperties.EVENT_TIME, String.valueOf(event.getEventTime()));

            rocketMsg.setKeys(event.getEventKey());
            rocketMsg.setWaitStoreMsgOK(true); //todo
//            log.debug("【领域事件】消息即将发送, topic->{}, message-key->{}", event.getTopic(), event.getEventKey());
            sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, producer.getSendMsgTimeout());
        } catch (Throwable e) {
            final String sendStatus = Optional.ofNullable(sendResult).map(x -> x.getSendStatus().name()).orElse("NONE");
            log.error(MessageFormat.format("【事件】消息发送失败:{0}, topic->{1}, msg-key->{2}, 发送状态:{3} ", e.getMessage(), event.getTopic(), event.getEventKey(), sendStatus), e);
            throw new RocketMQEventException(e);
        }

        long totalTime =  Math.max(System.currentTimeMillis() - event.getEventTime(),0);
        long publishTime = System.currentTimeMillis() - publishBegin;
        if (Objects.nonNull(sendResult) && sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            if (totalTime > 30) {
                log.warn("【事件】消息发送成功, 花费-> {} ms | {} ms,  topic->{}, msg-id->{}, msg-key->{}", publishTime, totalTime, event.getTopic(), sendResult.getMsgId(), event.getEventKey());
            } else {
                log.debug("【事件】消息发送成功, 花费-> {} ms | {} ms, topic->{}, msg-id->{}, msg-key->{}", publishTime, totalTime, event.getTopic(), sendResult.getMsgId(), event.getEventKey());
            }

            return;
        }

        final String sendStatus = Optional.ofNullable(sendResult).map(x -> x.getSendStatus().name()).orElse("NONE");
        log.error("【事件】消息发送失败, 花费->{} 毫秒, topic->{}, msg-key->{}, sendStatus->{}", totalTime, event.getTopic(), event.getEventKey(), sendStatus);

        throw new RocketMQEventException("发送失败" + sendStatus);
    }

    private Message createMessage(String topic, String tag, byte[] payload) {
        if (Objects.isNull(tag)) {
            tag = "";
        }
        return new Message(topic, tag, payload);
    }


}
